package com.zt.bigdata.flink.stream

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector

/**
  * @ClassName ApplyDemo
  * @Description
  * @Author zhangtonghy
  * @Date 2019-07-16 11:10
  * @Copyright: 版权所有 (C) zt zt.
  * @注意 ：本内容仅限于zt内部传阅，禁止外泄以及用于其他的商业目的
  **/
object ConnectDemo {


  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val left: DataStream[String] = env.socketTextStream("localhost", 999)
    val right: DataStream[Int] = env.socketTextStream("localhost", 998).map(_.toInt)


    val connect = left.connect(right)
      .process(new CoProcessFunction[String, Int, String] {

        override def open(parameters: Configuration): Unit = super.open(parameters)

        override def close(): Unit = super.close()

        override def processElement1(value: String, ctx: CoProcessFunction[String, Int, String]#Context, out: Collector[String]): Unit = {
          out.collect(value.toUpperCase)
        }

        override def processElement2(value: Int, ctx: CoProcessFunction[String, Int, String]#Context, out: Collector[String]): Unit = {
          out.collect((value * 10).toString)
        }
      })

    connect.print()

    env.execute()
  }
}
